package com.atguigu.medical.realtime.app.dwd;

import com.atguigu.medical.realtime.app.BaseSQLApp;
import com.atguigu.medical.realtime.common.Constant;
import com.atguigu.medical.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2023/5/26 18:07
 */
public class DwdTradePrescriptionPaySuccess extends BaseSQLApp {
    public static void main(String[] args) {
        new DwdTradePrescriptionPaySuccess().init(
            30003,
            2,
            "DwdTradePrescriptionPaySuccess"
        );
        
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        // 1. 读取 ods_medical
        readOdsMedical(tEnv, "DwdTradePrescriptionPaySuccess");
        // 2. 从ods_medical过滤支付表数据
        Table paymentPrescriptionTable = tEnv
            .sqlQuery("select " +
                          " after['id'] id,  " +
                          " after['update_time'] payment_time, " +
                          " after['user_id'] user_id, " +
                          " after['prescription_id'] prescription_id, " +
                          " after['payment_amount'] payment_amount," +
                          " ts " +
                          "from ods_medical " +
                          "where source['db']='medical' " +
                          "and source['table']='payment' " +
                          "and op='u' " +
                          "and before['status']='202' " +
                          "and after['status']='203' " +
                          "and after['prescription_id'] is not null ");
        tEnv.createTemporaryView("payment_prescription", paymentPrescriptionTable);
        // 3. 读取 dwd 层处方详情表
        tEnv.executeSql("create table dwd_diagnosis_prescription_detail(" +
                            " pre_detail_id string, " +
                            " prescription_id string, " +
                            " pre_detail_instruction string, " +
                            " medicine_id string, " +
                            " medicine_ct string, " +
                            " doctor_id string, " +
                            " patient_id string, " +
                            " consultation_id string, " +
                            " prescription_time string, " +
                            " total_amount string, " +
                            " split_amount string, " +
                            " ts bigint " +
                            ")" + SQLUtil.getKafkaSourceDDL(Constant.TOPIC_DWD_DIAGNOSIS_PRESCRIPTION_DETAIL, "DwdTradePrescriptionPaySuccess"));
        
        // 4. 两表 join
        Table result = tEnv
            .sqlQuery("select " +
                          " detail.pre_detail_id, " +
                          " pay.prescription_id, " +
                          " pay.payment_time, " +
                          " pay.user_id, " +
                          " detail.medicine_id, " +
                          " detail.doctor_id, " +
                          " detail.patient_id, " +
                          " detail.consultation_id, " +
                          " pay.payment_amount total_payment_amount, " +
                          " detail.split_amount split_payment_amount, " +
                          " pay.ts " +
                          "from payment_prescription pay " +
                          "join dwd_diagnosis_prescription_detail detail " +
                          "on pay.prescription_id=detail.prescription_id");
        
        // 5. 写出到 kakfa
        tEnv.executeSql("create table dwd_trade_prescription_success(" +
                            " pre_detail_id string, " +
                            " prescription_id string, " +
                            " payment_time string, " +
                            " user_id string, " +
                            " medicine_id string, " +
                            " doctor_id string, " +
                            " patient_id string, " +
                            " consultation_id string, " +
                            " total_payment_amount string, " +
                            " split_payment_amount string, " +
                            " ts bigint " +
                            ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_TRADE_PRESCRIPTION_SUCCESS));
        result.executeInsert("dwd_trade_prescription_success");
    
    }
}
